package hefei

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/4/20 0020.
  */
object TestConnectedComponents {
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    // Load the graph as in the PageRank example
    val graph = GraphLoader.edgeListFile(sc, "followers.txt")
    println("展示原图的顶点:")
    graph.vertices.collect.foreach(println(_))
    println("展示原图的边")
    graph.edges.collect.foreach(println(_))
    // Find the connected components
    val cc = graph.connectedComponents().vertices
    println("展示cc的顶点:")
    cc.collect.foreach(println(_))
    // Join the connected components with the usernames
    val users = sc.textFile("users.txt").map { line =>
      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }
    val ccByUsername = users.join(cc).map {
      case (id, (username, cc)) => (username, cc)
    }
    // Print the result
    println(ccByUsername.collect().mkString("\n"))
    sc.stop()
  }
}
